package org.amos.kafka.core;

import lombok.extern.slf4j.Slf4j;
import org.amos.kafka.callback.DefaultSendCallback;
import org.amos.kafka.callback.SendCallback;
import org.amos.kafka.executor.KafkaSenderExecutor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;

import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ForkJoinPool;

@Slf4j
public class KafkaSender {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Autowired(required = false)
    private KafkaSenderExecutor kafkaSenderExecutor;
    @Autowired
    private DefaultSendCallback defaultSendCallback;
    /**
     * 默认消息发送方法
     *
     * @param topic
     * @param msg
     */
    public void send(String topic, String msg) {
        //发送消息
        CompletableFuture<SendResult<String, Object>> completable = kafkaTemplate.send(topic, msg);
        commonHandler(topic, null, msg, completable, defaultSendCallback);
    }

    /**
     * 默认消息发送方法
     *
     * @param topic
     * @param key
     * @param msg
     */
    public void send(String topic, String key, String msg) {
        //发送消息
        CompletableFuture<SendResult<String, Object>> completable = kafkaTemplate.send(topic, key, msg);
        commonHandler(topic, key, msg, completable, defaultSendCallback);
    }

    /**
     * 默认消息发送方法,带回调消息
     * @param topic
     * @param key
     * @param msg
     * @param sendCallback
     */
    public void sendCallback(String topic, String key, String msg, SendCallback sendCallback) {
        //发送消息
        CompletableFuture<SendResult<String, Object>> completable = kafkaTemplate.send(topic, key, msg);
        commonHandler(topic, key, msg, completable, sendCallback);
    }

    /**
     * 消息发送至指定partition
     *
     * @param topic
     * @param partition
     * @param key
     * @param timestamp
     * @param msg
     */
    public void sendPartition(String topic, Integer partition, String key, Long timestamp, String msg) {
        //发送消息
        CompletableFuture<SendResult<String, Object>> completable = kafkaTemplate.send(topic, partition, timestamp, key, msg);
        commonHandler(topic, key, msg, completable, defaultSendCallback);
    }

    public void sendPartitionCallback(String topic, Integer partition, String key, Long timestamp, String msg, SendCallback sendCallback) {
        //发送消息
        CompletableFuture<SendResult<String, Object>> completable = kafkaTemplate.send(topic, partition, timestamp, key, msg);
        commonHandler(topic, key, msg, completable, sendCallback);
    }
    /**
     * 消息发送通用处理
     * @param topic
     * @param key
     * @param msg
     * @param completable
     * @param sendCallback
     */
    private void commonHandler(String topic, String key, String msg, CompletableFuture<SendResult<String, Object>> completable, SendCallback sendCallback) {

        completable.whenCompleteAsync((result, ex) -> {
            if (Objects.nonNull(ex)) {
                log.error("[KafkaSender][{}]msg:{},发送消息失败:{}", topic, msg, ex.getMessage());
                sendCallback.onFailure(result, ex);
                return;
            }
            // 根据业务系统需求自行设置，防止日志量过多浪费系统资源
            log.info("[KafkaSender][{}]发送消息成功:{}", topic, result.toString());
            if (Objects.nonNull(sendCallback)) {
                sendCallback.onSuccess(result);
            }
        }, Objects.nonNull(kafkaSenderExecutor) ? kafkaSenderExecutor : new ForkJoinPool());
    }
}
